/*******************************************************************************
 * Package: com.hadoop.mr
 * Type:    WordCountMain
 * Date:    2024-05-14 22:47
 *
 * Copyright (c) 2024 LTD All Rights Reserved.
 *
 * You may not use this file except in compliance with the License.
 *******************************************************************************/
package com.hadoop.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.net.URI;

/**
 * 功能描述： 远程读 远程写 数据写入到 hdfs
 * 写Hdfs文件服务器   https://blog.csdn.net/m0_72168501/article/details/128308292
 * @author Songxianyang
 * @date 2024-05-14 22:47
 */
public class WordCountMain {
    public static void main(String[] args) throws Exception{
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS","hdfs://localhost:9000");

        Job job = Job.getInstance(configuration);
        // 设置启动类
        job.setJarByClass(WordCountMain.class);

        // 设置 map   reduce
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReduce.class);

        // 设置 Combiner 减少开销 在map端先聚合一下 在传送给reduce
        job.setCombinerClass(WordCountCombiner.class);

        // map 输出 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        //Reduce 输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 创建输入目录
        Path inputPath = new Path("/hadooptest/input.txt");
        // 创建输出目录
        Path outputPath = new Path("/hadooptest/output/");
        // 获取FileSystem
        URI uri = new URI("hdfs://localhost:9000");
        FileSystem fs = FileSystem.get(uri, configuration, "SongXianYang");
        // 如果存在就删除
        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);
        }
        // 设置 FileInputFormat   FileOutputFormat
        FileInputFormat.setInputPaths(job,inputPath);
        FileOutputFormat.setOutputPath(job,outputPath);

        boolean result = job.waitForCompletion(true);
        RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(outputPath, true);
        while (iterator.hasNext()) {
            LocatedFileStatus next = iterator.next();
            FSDataInputStream inputStream = fs.open(next.getPath());
            IOUtils.copyBytes(inputStream, System.out, 4096, false);
        }
        System.exit(result ? 0 : 1);
    }
}
